Flink best practice: synchronizing MySQL data to TiDB using Canal

Background introduction

This article will introduce how to import the data in MySQL into Kafka in the form of Binlog + Canal, and then be consumed by Flink.

In order to quickly verify the functionality of the whole process, all components are deployed in a single machine. If you have insufficient physical resources, you can build all the components in this article in a 4G 1U virtual machine environment.

If you need to deploy in a production environment, it is recommended to replace each component with a highly available cluster deployment scheme.

Among them, we have created a set of Zookeeper single node environment, which is shared by Flink, Kafka, Canal and other components.

For all components requiring JRE, such as Flink, Kafka, Canal and Zookeeper, considering that upgrading JRE may affect other applications, we choose each component to use its own JRE environment independently.

This paper is divided into two parts. The first seven sections mainly introduce the construction of the basic environment, and the last section introduces how the data flows in each component.

Data flows through the following components:

Generating Binlog from MySQL data source

The Canal reads the Binlog, generates the Canal json, and pushes it to the Topic specified by Kafka

Flink uses the Flink SQL connector Kafka API to consume data in the Kafka Topic

Flink writes data to TiDB through the Flink connector JDBC

The structure of TiDB + Flink supports the development and running of many different kinds of applications.

At present, the main features include:

Batch flow integration

Sophisticated state management

Event time support

Accurate primary state consistency guarantee

Flink can run on a variety of resource management frameworks including YARN, Mesos and Kubernetes. It also supports independent deployment on bare metal clusters. TiDB can be deployed on AWS, Kubernetes, GCP and gke. It also supports independent deployment on bare metal clusters using TiUP.

The common applications of TiDB + Flink structure are as follows:

Event driven applications

Anti fraud

anomaly detection

Rule based alarm

Business process monitoring

Data analysis application

Network quality monitoring

Product update and test evaluation analysis

Impromptu analysis of factual data

Large scale graph analysis

Data pipeline application

Construction of e-commerce real-time query index

E-commerce continuous ETL

Operating system environment

software environment

Machine allocation

Compared with the traditional stand-alone database, TiDB has the following advantages:

Pure distributed architecture, with good scalability and supporting elastic capacity expansion and contraction

It supports SQL, exposes the network protocol of MySQL, and is compatible with most MySQL syntax. It can directly replace MySQL in most scenarios

High availability is supported by default. When a few replicas fail, the database itself can automatically perform data repair and failover, which is transparent to the business

It supports ACID transactions and is friendly to some scenarios with strong consistent requirements, such as bank transfer

It has rich tool chain ecology, covering a variety of scenarios such as data migration, synchronization and backup

In terms of kernel design, TiDB distributed database divides the overall architecture into multiple modules, and each module communicates with each other to form a complete TiDB system. The corresponding architecture diagram is as follows:

In this article, we only do the simplest function test, so we deploy a set of single node but replica TiDB, which involves the following three modules:

TiDB Server: SQL layer, which exposes the connection endpoint of MySQL protocol. It is responsible for accepting client connections, performing SQL parsing and optimization, and finally generating distributed execution plans.

PD (Placement Driver) Server: the meta information management module of the whole TiDB cluster, which is responsible for storing the real-time data distribution of each TiKV node and the overall topology of the cluster, providing the TiDB Dashboard control interface, and assigning transaction ID s to distributed transactions.

TiKV Server: it is responsible for storing data. Externally, TiKV is a distributed key value storage engine that provides transactions.

TiUP deployment template file

TiDB Cluster environment

The focus of this article is not to deploy TiDB Cluster. As a rapid experimental environment, TiDB Cluster cluster with a single copy is deployed on only one machine. There is no need to deploy the monitoring environment.

Create a table for testing

In this experiment, Zookeeper environment is configured separately to provide services for Kafka and Flink environments.

As an experimental demonstration scheme, only the stand-alone environment is deployed.

Unzip the Zookeeper package

Deploy jre for Zookeeper

Modify / opt/zookeeper/bin/zkEnv.sh file to add JAVA_HOME environment variable

Create a profile for Zookeeper

Start Zookeeper

Check the status of Zookeeper

Suggestions on Zookeeper

I personally have an immature suggestion about Zookeeper:

The Zookeeper cluster version must enable network monitoring.

In particular, pay attention to the network bandwidth in system metrics.

Kafka is a distributed stream processing platform, which is mainly used in two types of applications:

A real-time stream data pipeline is constructed, which can reliably obtain data between systems or applications( Equivalent to message queue)

Build real-time streaming applications to transform or influence these streaming data( That is, flow processing, which changes internally between kafka stream topic and topic)

Kafka has four core API s:

The Producer API allows an application to publish a stream of data to one or more Kafka topic s.

The Consumer API allows an application to subscribe to one or more topic s and process the streaming data published to them.

The Streams API allows an application as a stream processor to consume input streams generated by one or more topics, and then produce an output stream to one or more topics for effective conversion in the input and output streams.

The Connector API allows to build and run reusable producers or consumers to connect Kafka topics to existing applications or data systems. For example, connect to a relational database and capture all changes to the table.

In this experiment, only functional verification is done, and only a stand-alone Kafka environment is built.

Download and unzip Kafka

Modify Kafka's jre environment variable

Modify Kafka profile

Start Kafka

View version information of Kafka

Where 2.13 is the version information of scale and 2.7.0 is the version information of Kafka.

Apache Flink is a framework and distributed processing engine for stateful computing on unbounded and bounded data streams. Flink can run in all common cluster environments and can calculate at memory speed and any size.

Apache Flink, a distributed processing framework supporting high throughput, low latency and high performance, is a framework and distributed processing engine for stateful computing of unbounded and bounded data streams. Flink is designed to run in all common cluster environments and perform calculations at memory execution speed and any size.

This experiment only does functional testing and only deploys the stand-alone Flink environment.

Download and distribute Flink

Deploy jre for Flink

Add lib required by Flink

Modify Flink profile

Launch Flink

View Flink GUI

Unzip MySQL package

Create MySQL Service file

Create my.cnf file

Initialize and start MySQL

Create a new MySQL user to connect to Canal

Create tables for testing in MySQL

The main purpose of Canal is to provide incremental data subscription and consumption based on MySQL database incremental log parsing

In the early days, Alibaba had the business requirement of cross machine room synchronization due to the deployment of dual machine rooms in Hangzhou and the United States. The implementation method was mainly to obtain incremental changes based on the business trigger.

Since 2010, the business has gradually tried to obtain incremental changes through database log parsing for synchronization, resulting in a large number of database incremental subscriptions and consumption services.

Businesses based on log incremental subscription and consumption include:

database mirroring

Real time database backup

Index construction and real-time maintenance (splitting heterogeneous indexes, inverted indexes, etc.)

Service cache refresh

Incremental data processing with business logic

Current canal supports source MySQL versions, including 5.1. X, 5.5. X, 5.6. X, 5.7. X, and 8.0. X

Unzip the Canal package

Deploy jre of Canal

Modify the configuration file of Canal

MySQL binlog - > canal - > Kafka channel

View MySQL Binlog information

Check the MySQL Binlog information to ensure that the Binlog is normal

Create a Topic in Kafka

Create a topic Canal Kafka in Kafka. The name of this topic should correspond to Canal. MQ. Topic = Canal Kafka in the Canal configuration file / opt/canal/conf/example/instance.properties:

View all topics in Kafka:

View the information of topic ticdc test in Kafka:

8.1.3 start Canal

Before starting the Canal, you need to check the port on the Canal node:

Start Canal:

View Canal log

View / opt/canal/logs/example/example.log

View consumer information in Kafka

Insert a test message into MySQL:

Check the information of the consumer, and you have the test data just inserted:

Kafka - > Flink path

Create the t2 table in Flink, and the connector type is kafka

Insert a test data in MySQL:

Sync data in real time from Flink:

Flink - > tidb path

Create a table for testing in the downstream TiDB

Create a test table in Flink

View the inserted data in the downstream TiDB